package cn.tedu.charging.order.rabbitmq;

import com.rabbitmq.client.AMQP;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.ConfirmListener;
import lombok.extern.slf4j.Slf4j;
import org.junit.jupiter.api.Test;

import java.io.IOException;
import java.nio.charset.StandardCharsets;
import java.util.HashMap;
import java.util.Map;
@Slf4j
public class Producer {
    //测试生产发送
    @Test
    public void send2Queue01() throws IOException {
        //1.拿到连接对象channel
        Channel channel = ChannelManager.createChannel();
        //2.组织一个消息对象
        //2.1消息体 是数据字节数组 所以字符串 对象转化成byte数组
        byte[] body="hello rabbitmq".getBytes(StandardCharsets.UTF_8);
        //2.2消息头
        //固定头 是rabbitmq给提供的属性
        AMQP.BasicProperties.Builder propsBuilder=new AMQP.BasicProperties.Builder();
        //每个属性都有自己的固定名字和固定作用
        propsBuilder.appId("appId");//生产发送消息的来运项目软件的id
        propsBuilder.type("testMsg");//自定义的消息类型
        //自定义 随便携带的头 是一个Map<String,Object>对象
        Map<String,Object> headers=new HashMap<>();
        headers.put("name","王翠花");
        propsBuilder.headers( headers);
        AMQP.BasicProperties props = propsBuilder.build();
        //2.3发送消息之后 消息能够到达哪个队列,可以由路由key决定
        String routingKey="queue01";//当前案例使用的路由key就是刚刚声明的队列名称
        //3.将消息发送出去
        //参数1 String exchange: 交换机名称
        channel.basicPublish("amq.fanout",routingKey,props,body);
    }

    @Test
    public void sendSync() throws Exception {
        //拿到channel
        Channel channel = ChannelManager.createChannel();
        //记录执行发送其实时间 时间戳长整型
        Long start=System.currentTimeMillis();
        //1.开启发送确认
        channel.confirmSelect();
        //2.执行发送
        channel.basicPublish("","queue1",null,"send sync message".getBytes());
        //3.阻塞线程 直到rabbitmq反馈消息接收确认结果
        boolean ack = channel.waitForConfirms();
        if (ack){
            log.info("消息发送成功");
        }else{
            log.info("消息发送失败");
        }
        //记录执行发送的结束时间
        Long end=System.currentTimeMillis();
        log.info("耗时:{}MS",end-start);
    }
    @Test
    public void sendAsync() throws Exception {
        //拿到channel
        Channel channel=ChannelManager.createChannel();
        Long start = System.currentTimeMillis();
        //发送消息异步确认 提供异步监听回调对象 接收异步确认结果
        channel.confirmSelect();
        channel.addConfirmListener(new ConfirmListener() {
            //收到异步回调 消息成功
            @Override
            public void handleAck(long deliveryTag, boolean multiple) throws IOException {
                log.info("消息发送成功");
            }
            //收到异步回到 消息失败
            @Override
            public void handleNack(long deliveryTag, boolean multiple) throws IOException {
                log.info("消息发送失败");
            }
        });
        channel.basicPublish("","queue1",null,"send sync message".getBytes());
        Long end=System.currentTimeMillis();
        log.info("耗时:{}MS",end-start);
        //让主线程阻塞一下,否则有可能主线程执行完成方法 回调还没有看到确认结果
        //Thread.sleep(100000);
    }
    //发送一个消息超时的测试
    @Test
    public void sendTtl() throws IOException {
        Channel channel = ChannelManager.createChannel();
        //1.组织消息 带有超时属性
        byte[] body="ttl message".getBytes(StandardCharsets.UTF_8);
        AMQP.BasicProperties.Builder propsBuilder=new AMQP.BasicProperties.Builder();
        //expiration
        propsBuilder.expiration("50000");
        AMQP.BasicProperties props = propsBuilder.build();
        //2.发送消息
        channel.basicPublish("","queue3",props,body);
    }
}
